package cn.jly.bigdata.spark.sql

import java.util.Properties

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

/**
 * @author lanyangji
 * @date 2019/12/2 13:50
 */
object SparkSql08_Mysql {

  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .config(new SparkConf())
      .master("local[*]")
      .appName("SparkSql08_Mysql")
      .getOrCreate()

    println("---------从mysql中读取数据创建 dataFrame 方法一，这是通用方式----------------")
    val df: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql:///rdd")
      .option("dbtable", "user")
      .option("user", "root")
      .option("password", "123456")
      .load()

    df.show()

    df.createOrReplaceTempView("user")
    spark.sql("SELECT name FROM user WHERE id IN(2,3,4)").show()

    println("---------从mysql中读取数据创建 dataFrame 方法二，这是简写方式----------------")
    val connectionProperties: Properties = new Properties()
    connectionProperties.put("user", "root")
    connectionProperties.put("password", "123456")

    val df2: DataFrame = spark.read.jdbc(
      "jdbc:mysql:///rdd",
      "user",
      connectionProperties
    )

    df2.show()

    println("----------------将数据写入到Mysql方法一，这是通用方式-------------------")

    df2
      .write
      .format("jdbc")
      .mode(SaveMode.Overwrite)
      .option("url", "jdbc:mysql:///rdd")
      .option("dbtable", "user2")
      .option("user", "root")
      .option("password", "123456")
      .save()

    println("----------------将数据写入到Mysql方法二，这是简写方式-------------------")
    df2
      .write
      .mode(SaveMode.Overwrite)
      .jdbc(
        "jdbc:mysql:///rdd",
        "user3",
        connectionProperties
      )

    // 释放资源
    spark.close()
  }
}
